package com.liunian.proxy.handler;

import com.liunian.NettyBootstrapInitializer;
import com.liunian.WrpcBootstrap;
import com.liunian.annotation.TryTimes;
import com.liunian.compress.CompressorFactory;
import com.liunian.discovery.Registry;
import com.liunian.exceptions.DiscoveryException;
import com.liunian.exceptions.NetworkException;
import com.liunian.protection.CircuitBreaker;
import com.liunian.serialize.SerializerFactory;
import com.liunian.transport.message.RequestPayload;
import com.liunian.transport.message.WrpcRequest;
import com.liunian.wrpcenum.RequestType;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import lombok.extern.slf4j.Slf4j;

import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.util.Map;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * 该类封装了客户端通信的基础逻辑，每一个代理对象的远程调用过程都封装在了invoke方法中
 * 1、发现可用服务
 * 2、建立连接
 * 3、发送请求
 * 4、得到结果
 */
@Slf4j
public class RpcConsumerInvocationHandler implements InvocationHandler {
    // 此处需要一个注册中心和一个接口
    private final Registry registry;
    private final Class<?> interfaceRef;
    private String group;

    public RpcConsumerInvocationHandler(Registry registry, Class<?> interfaceRef, String group) {
        this.registry = registry;
        this.interfaceRef = interfaceRef;
        this.group = group;
    }

    /**
     * 所有的方法调用本质都会走到这里
     *
     * @param proxy
     * @param method
     * @param args
     * @return
     * @throws Throwable
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        // 从接口中判断是否需要重试
        TryTimes tryTimesAnnotation = method.getAnnotation(TryTimes.class);
        // 设置重试默认值 表示不重试
        int tryTimes = 0;
        int intervalTime = 0;
        if (tryTimesAnnotation != null) {
            tryTimes = tryTimesAnnotation.tryTimes();
            intervalTime = tryTimesAnnotation.intervalTime();
        }
        while (true) {
            /**
             * --------------------------------------封装报文-----------------------------------
             */
            // 一、封装报文
            RequestPayload requestPayload = RequestPayload.builder()
                    .interfaceName(interfaceRef.getName())
                    .methodName(method.getName())
                    .parametersType(method.getParameterTypes())
                    .parametersValue(args)
                    .returnType(method.getReturnType())
                    .build();
            // 创建一个请求
            WrpcRequest wrpcRequest = WrpcRequest.builder()
                    .requestId(WrpcBootstrap.getInstance().getConfiguration().getIdGenerator().getId())
                    .compressType(CompressorFactory.getCompressor(WrpcBootstrap.getInstance().getConfiguration().getCompressType()).getCode())
                    .requestType(RequestType.REQUEST.getId())
                    .serializeType(SerializerFactory.getSerializer(WrpcBootstrap.getInstance().getConfiguration().getSerializeType()).getCode())
                    .timeStamp(System.currentTimeMillis())
                    .requestPayload(requestPayload)
                    .build();
            // 2、将请求存入本地线程，需要在合适的时候调用remove方法
            WrpcBootstrap.REQUEST_THREAD_LOCAL.set(wrpcRequest);


            // method 具体的方法，args参数列表
//        log.info("method --> {}", method.getName());
//        log.info("args --> {}", args);
            // 3、、发现服务，从注册中心寻找一个可用的服务
//      InetSocketAddress address = registry.lookup(interfaceRef.getName());
            // 传入服务的名字。返回一个IP+Port
            // 我们每次调用相关方法的时候都需要去注册中心拉取服务列表吗？
            //         我们如何合理的选择一个可用的服务，而不是只获取第一个

            // 二、获取当前配置的负载均衡器，选取一个可用节点
            InetSocketAddress address = WrpcBootstrap.getInstance().getConfiguration().getLoadBalancer().selectServiceAddress(interfaceRef.getName(),group);
            if (log.isDebugEnabled()) {
                log.debug("服务调用方发现了[{}]的可用主机[{}]", interfaceRef.getName(), address);
            }

            // 什么情况下需要重试， 1、发生异常 2、响应有问题 code=500
            // 4、获取当前地址所对应的断路器，如果断路器是打开的，就不发送请求了，直接抛异常。
            Map<SocketAddress, CircuitBreaker> everyIpCircuitBreaker = WrpcBootstrap.getInstance().getConfiguration().getEveryIpCircuitBreaker();
            CircuitBreaker circuitBreaker = everyIpCircuitBreaker.get(address);
            if (circuitBreaker == null) {
                circuitBreaker = new CircuitBreaker(10,0.5F);
                everyIpCircuitBreaker.put(address,circuitBreaker);
            }
            try {
                // 如果不是心跳请求并且断路器是打开的，就不响应了，直接抛异常。
                if (wrpcRequest.getRequestType() != RequestType.HEART_BEAT.getId() && circuitBreaker.isBreak()) {
                    // 断路器打开，说明当前的请求不应该再发送，
                    Timer timer = new Timer();
                    // 如果断路器打开，3秒后重置，再发送请求。
                    timer.schedule(new TimerTask() {
                        @Override
                        public void run() {
                            WrpcBootstrap.getInstance().getConfiguration().getEveryIpCircuitBreaker()
                                    .get(address).reset();
                        }
                    }, 5000);
                    throw new RuntimeException("当前断路器已经开启，无法发送请求被");
                }

                // 使用netty连接服务器，发送调用的服务的名字+方法名字+参数列表，得到结果
                // 定义线程池。 EventLoopGroup
                // q:整个连接过程放在这里行不行，放在这里也就意味着，每次调用都会产生一个新的netty连接。 所以在这里我们需要考虑的问题是如何缓存我们的连接。
                // 也就意味着，每次在此处建立一个新的连接是不合适的

                // 解决方案？缓存channel，尝试从缓存中获取channel，如未获取，则创建新的连接并缓存。
                // 5、尝试获取一个可用的通道
                Channel channel = getAvailableChannel(address);
                if (log.isDebugEnabled()) {
                    log.debug("获取了和[{}]建立的连接通道，准备发送数据。", address);
                }

                /**
                 * ------------------------------------同步策略-------------------------------------
                 */
//                // writeAndFlush 发送数据
//                ChannelFuture channelFuture = channel.writeAndFlush(new Object()).await();
//                // 需要学习channelFuture的一些简单的api
//                // get() 阻塞获取结果
//                // getNow() 获取当前的结果，如果未处理完成返回null。
//                /**
//                 * isDone()是否执行完成
//                 * isSuccess() 是否执行成功
//                 */
//                if (channelFuture.isDone()) {
//                    Object object = channelFuture.getNow();
//                } else if (!channelFuture.isSuccess()) {
//                    // 需要捕获或异常，可以捕获异步任务中的异常。
//                    Throwable cause = channelFuture.cause();
//                    throw new RuntimeException(cause);
//                }
                /**
                 * ------------------------------------异步策略-------------------------------------
                 */
                // 6、写出报文
                CompletableFuture<Object> completableFuture = new CompletableFuture<>();
                // 将 completableFuture 暴露出去
                WrpcBootstrap.PENDING_REQUEST.put(wrpcRequest.getRequestId(), completableFuture);
                // 这里直接 writeAndFlush 写出了一个请求，这个请求的实例就会进入pipeline执行出站的一些列操作
                // 我们可以想象得到，第一个出站程序一定是将这个请求对象 wrpcRequest 转化成一个二进制的报文。
                channel.writeAndFlush(wrpcRequest).addListener((ChannelFutureListener) promise -> {
                    // 当前的promise将来返回的结果是什么？writeAndFlush的返回结果
                    // 一旦数据被写出去，这个promise也就结束了
                    // 但是我们想要的是什么？ 我们想要的是服务端给我们的返回值，所以这里处理的completableFuture是有问题的。
                    // 是不是应该将 completableFuture 挂起并且暴露，并且在得到服务提供方的响应 的时候调用complete()方法。
//                    if (promise.isDone()) {
//                        completableFuture.complete(promise.getNow());
//                    }
                    // 只需处理一下异常就可以了
                    if (!promise.isSuccess()) {
                        completableFuture.completeExceptionally(promise.cause());
                    }
                });
//                Object o = completableFuture.get(3, TimeUnit.SECONDS);

                // 7、清理threadLocal
                WrpcBootstrap.REQUEST_THREAD_LOCAL.remove();

                // 将结果返回
                // 如果没有地方处理这个 completableFuture ，这里会阻塞，等待complete()方法的执行
                // q: 我们需要在哪里调用complete方法得到结果，很明显 pipeline 中最终的handler的处理结果
                // 8、获得响应的结果
                Object result = completableFuture.get(10, TimeUnit.SECONDS);
                // 记录成功的请求。
                circuitBreaker.recordRequest();
                return result;
            } catch (Exception e) {
                // 如果捕获到异常，次数减一，并且等待固定时间，但是固定时间有一定的问题，可能导致充实风暴。
                tryTimes--;
                // 记录错误的次数
                circuitBreaker.recordErrorRequest();
                try {
                    Thread.sleep(intervalTime);
                } catch (InterruptedException ex) {
                    log.error("重试时发生打断异常。", ex);
                }
                if (tryTimes < 0) {
                    log.error("对方法[{}]进行远程调用时，重试[{}]次，仍不可调用。。", method.getName(), tryTimes, e);
                    break;
                }
                log.error("进行第[{}]次重试时发生异常。", 3 - tryTimes, e);
            }
        }
        throw new RuntimeException("执行远程方法" + method.getName() + "调用失败。");
    }

    /**
     * 根据地址获取一个可用通道channel
     *
     * @param address
     * @return
     */
    private Channel getAvailableChannel(InetSocketAddress address) {
        // 1、尝试从缓存中获取
        Channel channel = WrpcBootstrap.CHANNEL_CACHE.get(address);
        // 2、如果缓存中获取不到就去建立连接
        if (channel == null) {
            // await方法会阻塞，会等待连接成功再返回，netty还提供了异步处理的逻辑。
            // sync 和 await 都是阻塞当前线程获取返回值。 (因为连接的过程是异步的，发送数据的过程是异步的。)
            // 如果发生了异常sync会主动在主线程抛出异常;await不会，它的异常在子线程中处理需要使用future中处理。
//                    channel = NettyBootstrapInitializer.getBootstrap()
//                            .connect(address).await().channel();
            // 使用addListener执行的异步操作
            CompletableFuture<Channel> channelFuture = new CompletableFuture<>();
            NettyBootstrapInitializer.getBootstrap().connect(address).addListener(
                    (ChannelFutureListener) promise -> {
                        if (promise.isDone()) {
                            // 异步的,我们已经完成
                            if (log.isDebugEnabled()) {
                                log.debug("已经和[{}]成功建立了连接。", address);
                            }
                            channelFuture.complete(promise.channel());
                        } else if (!promise.isSuccess()) {
                            channelFuture.completeExceptionally(promise.cause());
                        }
                    });
            // 阻塞获取channel
            try {
                channel = channelFuture.get(3, TimeUnit.SECONDS);
            } catch (InterruptedException | ExecutionException | TimeoutException e) {
                log.error("获取通道时发生异常。", e);
                throw new DiscoveryException(e);
            }
            // 缓存channel
            WrpcBootstrap.CHANNEL_CACHE.put(address, channel);
        }
        // 3、缓存中没拿到，创建的时候也出现问题，为了防止空指针异常，在这里进行一步判断
        if (channel == null) {
            log.error("获取或建立与[{}]通道时发生异常。", address);
            throw new NetworkException("获取通道时发生异常。");
        }
        return channel;
    }
}
